package raft.cmd;

import raft.core.RaftNode;
import raft.core.RaftState;
import raft.message.InMemoryMessageHandler;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * Raft Leader故障转移演示
 * 演示在日志同步过程中leader掉线后的重新选举和日志同步过程
 */
public class RaftFailoverDemo {
    
    public static void main(String[] args) throws InterruptedException {
        System.out.println("=== Raft Leader故障转移演示 ===");
        
        // 演示基本故障转移
        demonstrateLeaderFailover();
        
        Thread.sleep(2000);
        
        // 演示日志同步中的故障转移
        demonstrateFailoverDuringLogSync();
        
        Thread.sleep(2000);
        
        // 演示复杂场景：故障leader重新加入和任期冲突
        demonstrateComplexFailoverScenario();
    }
    
    /**
     * 演示基本的Leader故障转移
     */
    public static void demonstrateLeaderFailover() throws InterruptedException {
        System.out.println("\n=== 基本故障转移演示 ===");
        
        // 创建5个节点的集群以更好地演示故障转移
        List<String> nodeIds = Arrays.asList("node1", "node2", "node3", "node4", "node5");
        RaftNode[] nodes = new RaftNode[5];
        
        // 初始化节点
        for (int i = 0; i < 5; i++) {
            String nodeId = nodeIds.get(i);
            List<String> peers = nodeIds.stream()
                    .filter(id -> !id.equals(nodeId))
                    .collect(java.util.stream.Collectors.toList());
            
            nodes[i] = new RaftNode(nodeId, peers, new InMemoryMessageHandler());
            InMemoryMessageHandler.registerNode(nodeId, nodes[i]);
        }
        
        System.out.println("所有节点已启动，等待首次选举...");
        Thread.sleep(3000); // 等待选举完成
        
        printClusterStatus(nodes);
        
        // 找到当前leader
        RaftNode currentLeader = findLeader(nodes);
        if (currentLeader != null) {
            System.out.println("\n当前领导者: " + currentLeader.getNodeId());
            
            // 添加一些日志
            System.out.println("添加初始日志条目...");
            for (int i = 1; i <= 3; i++) {
                currentLeader.appendEntry("Initial-Command-" + i);
                Thread.sleep(200);
            }
            
            printClusterStatus(nodes);
            
            // 模拟leader故障
            System.out.println("\n模拟领导者 " + currentLeader.getNodeId() + " 故障...");
            String failedLeaderId = currentLeader.getNodeId();
            currentLeader.shutdown();
            InMemoryMessageHandler.unregisterNode(failedLeaderId);
            
            System.out.println("等待重新选举...");
            Thread.sleep(4000); // 等待重新选举（包含随机超时时间）
            
            // 检查新的集群状态
            System.out.println("\n重新选举后的集群状态:");
            RaftNode newLeader = null;
            for (RaftNode node : nodes) {
                if (!node.getNodeId().equals(failedLeaderId)) {
                    System.out.println("  " + node.getStatus());
                    if (node.getState() == RaftState.LEADER) {
                        newLeader = node;
                    }
                }
            }
            
            if (newLeader != null) {
                System.out.println("\n✓ 新领导者选出: " + newLeader.getNodeId());
                
                // 新leader添加更多日志
                System.out.println("新领导者继续添加日志...");
                for (int i = 4; i <= 6; i++) {
                    newLeader.appendEntry("Recovery-Command-" + i);
                    Thread.sleep(200);
                }
                
                Thread.sleep(1000);
                System.out.println("\n最终集群状态:");
                for (RaftNode node : nodes) {
                    if (!node.getNodeId().equals(failedLeaderId)) {
                        System.out.println("  " + node.getStatus());
                    }
                }
            } else {
                System.out.println("✗ 未能选出新领导者");
            }
        }
        
        // 清理剩余节点
        for (RaftNode node : nodes) {
            if (!node.getNodeId().equals(currentLeader.getNodeId())) {
                node.shutdown();
                InMemoryMessageHandler.unregisterNode(node.getNodeId());
            }
        }
    }
    
    /**
     * 演示在日志同步过程中的故障转移
     */
    public static void demonstrateFailoverDuringLogSync() throws InterruptedException {
        System.out.println("\n=== 日志同步中的故障转移演示 ===");
        
        // 创建3个节点
        RaftNode node1 = new RaftNode("sync1 ", Arrays.asList("sync2", "sync3"), new InMemoryMessageHandler());
        RaftNode node2 = new RaftNode("sync2", Arrays.asList("sync1", "sync3"), new InMemoryMessageHandler());
        RaftNode node3 = new RaftNode("sync3", Arrays.asList("sync1", "sync2"), new InMemoryMessageHandler());
        
        InMemoryMessageHandler.registerNode("sync1", node1);
        InMemoryMessageHandler.registerNode("sync2", node2);
        InMemoryMessageHandler.registerNode("sync3", node3);
        
        System.out.println("等待初始选举...");
        Thread.sleep(2000);
        
        RaftNode leader = findLeader(node1, node2, node3);
        if (leader != null) {
            System.out.println("初始领导者: " + leader.getNodeId());
            printClusterStatus(node1, node2, node3);
            
            // 开始异步添加大量日志，模拟日志同步过程
            System.out.println("\n开始批量日志同步...");
            CompletableFuture<Void> logSync = CompletableFuture.runAsync(() -> {
                try {
                    for (int i = 1; i <= 10; i++) {
                        if (leader.getState() == RaftState.LEADER) {
                            leader.appendEntry("Batch-Log-" + i);
                            System.out.println("  添加日志: Batch-Log-" + i);
                            Thread.sleep(300); // 模拟一定的同步时间
                        } else {
                            System.out.println("  领导者已不可用，停止添加日志");
                            break;
                        }
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
            
            // 在日志同步过程中让leader故障
            Thread.sleep(1500); // 让一些日志先同步
            
            System.out.println("\n在日志同步过程中模拟领导者故障...");
            String failedLeaderId = leader.getNodeId();
            leader.shutdown();
            InMemoryMessageHandler.unregisterNode(failedLeaderId);
            
            // 等待故障转移完成
            System.out.println("等待故障转移和重新选举（包含随机超时）...");
            Thread.sleep(4000); // 足够的时间让选举完成
            
            // 检查剩余节点状态
            System.out.println("\n故障转移后的状态:");
            RaftNode[] remainingNodes = {node1, node2, node3};
            RaftNode newLeader = null;
            
            for (RaftNode node : remainingNodes) {
                if (!node.getNodeId().equals(failedLeaderId)) {
                    System.out.println("  " + node.getStatus());
                    if (node.getState() == RaftState.LEADER) {
                        newLeader = node;
                    }
                }
            }
            
            if (newLeader != null) {
                System.out.println("\n✓ 新领导者: " + newLeader.getNodeId());
                
                // 新leader继续完成剩余的日志同步
                System.out.println("新领导者继续日志同步...");
                for (int i = 11; i <= 15; i++) {
                    newLeader.appendEntry("Recovery-Log-" + i);
                    Thread.sleep(200);
                }
                
                Thread.sleep(1000);
                System.out.println("\n最终同步状态:");
                for (RaftNode node : remainingNodes) {
                    if (!node.getNodeId().equals(failedLeaderId)) {
                        System.out.println("  " + node.getStatus());
                    }
                }
                
                // 验证日志一致性
                verifyLogConsistency(remainingNodes, failedLeaderId);
            }
            
            // 等待异步任务完成
            try {
                logSync.get(1, TimeUnit.SECONDS);
            } catch (Exception e) {
                // 预期会超时或被中断
            }
        }
        
        // 清理
        if (!node1.getNodeId().equals(leader.getNodeId())) {
            node1.shutdown();
            InMemoryMessageHandler.unregisterNode("sync1");
        }
        if (!node2.getNodeId().equals(leader.getNodeId())) {
            node2.shutdown();
            InMemoryMessageHandler.unregisterNode("sync2");
        }
        if (!node3.getNodeId().equals(leader.getNodeId())) {
            node3.shutdown();
            InMemoryMessageHandler.unregisterNode("sync3");
        }
    }
    
    /**
     * 验证日志一致性
     */
    private static void verifyLogConsistency(RaftNode[] nodes, String failedNodeId) {
        System.out.println("\n=== 验证日志一致性 ===");
        
        int[] logSizes = new int[nodes.length];
        int[] commitIndexes = new int[nodes.length];
        int activeNodes = 0;
        
        for (int i = 0; i < nodes.length; i++) {
            if (!nodes[i].getNodeId().equals(failedNodeId)) {
                logSizes[activeNodes] = nodes[i].getLogSize();
                commitIndexes[activeNodes] = nodes[i].getCommitIndex();
                activeNodes++;
            }
        }
        
        // 检查日志大小一致性
        boolean logSizeConsistent = true;
        for (int i = 1; i < activeNodes; i++) {
            if (logSizes[i] != logSizes[0]) {
                logSizeConsistent = false;
                break;
            }
        }
        
        // 检查提交索引一致性
        boolean commitConsistent = true;
        for (int i = 1; i < activeNodes; i++) {
            if (commitIndexes[i] != commitIndexes[0]) {
                commitConsistent = false;
                break;
            }
        }
        
        if (logSizeConsistent && commitConsistent) {
            System.out.println("✓ 日志一致性验证通过");
            System.out.println("  - 所有节点日志大小: " + logSizes[0]);
            System.out.println("  - 所有节点提交索引: " + commitIndexes[0]);
        } else {
            System.out.println("✗ 日志一致性验证失败");
            for (int i = 0; i < activeNodes; i++) {
                System.out.println("  - 节点日志大小: " + logSizes[i] + ", 提交索引: " + commitIndexes[i]);
            }
        }
    }
    
    private static void printClusterStatus(RaftNode... nodes) {
        System.out.println("\n--- 集群状态 ---");
        for (RaftNode node : nodes) {
            String status = node.getStatus();
            String role = "";
            switch (node.getState()) {
                case LEADER:
                    role = " (LEADER)";
                    break;
                case CANDIDATE:
                    role = " (CANDIDATE)";
                    break;
                case FOLLOWER:
                    role = " (FOLLOWER)";
                    break;
            }
            System.out.println("  " + status + role);
        }
        System.out.println("-------------------");
    }
    
    private static RaftNode findLeader(RaftNode... nodes) {
        for (RaftNode node : nodes) {
            if (node.getState() == RaftState.LEADER) {
                return node;
            }
        }
        return null;
    }
    
    /**
     * 演示复杂故障转移场景：故障leader重新加入和任期冲突处理
     */
    public static void demonstrateComplexFailoverScenario() throws InterruptedException {
        System.out.println("\n=== 复杂故障转移场景演示 ===");
        System.out.println("场景：故障leader重新加入 + 两个相同任期leader冲突");
        
        // 创建5个节点的集群
        List<String> nodeIds = Arrays.asList("complex1", "complex2", "complex3", "complex4", "complex5");
        RaftNode[] nodes = new RaftNode[5];
        
        // 初始化节点
        for (int i = 0; i < 5; i++) {
            String nodeId = nodeIds.get(i);
            List<String> peers = nodeIds.stream()
                    .filter(id -> !id.equals(nodeId))
                    .collect(java.util.stream.Collectors.toList());
            
            nodes[i] = new RaftNode(nodeId, peers, new InMemoryMessageHandler());
            InMemoryMessageHandler.registerNode(nodeId, nodes[i]);
        }
        
        System.out.println("\n第一阶段：初始集群启动和选举");
        Thread.sleep(3000); // 等待初始选举
        
        printClusterStatus(nodes);
        
        // 找到初始leader
        RaftNode initialLeader = findLeader(nodes);
        if (initialLeader == null) {
            System.out.println("✗ 初始选举失败！");
            return;
        }
        
        System.out.println("\n初始领导者: " + initialLeader.getNodeId() + ", 任期: " + initialLeader.getCurrentTerm());
        
        // 添加一些初始日志
        System.out.println("\n第二阶段：添加初始日志");
        for (int i = 1; i <= 3; i++) {
            initialLeader.appendEntry("Initial-Log-" + i);
            Thread.sleep(200);
        }
        
        printClusterStatus(nodes);
        
        // 模拟leader故障
        System.out.println("\n第三阶段：模拟领导者故障");
        String failedLeaderId = initialLeader.getNodeId();
        int failedLeaderTerm = initialLeader.getCurrentTerm();
        
        System.out.println("让领导者 " + failedLeaderId + " 故障...");
        initialLeader.shutdown();
        InMemoryMessageHandler.unregisterNode(failedLeaderId);
        
        // 等待新的选举
        System.out.println("等待重新选举...");
        Thread.sleep(4000);
        
        // 找到新leader
        RaftNode newLeader = null;
        for (RaftNode node : nodes) {
            if (!node.getNodeId().equals(failedLeaderId) && node.getState() == RaftState.LEADER) {
                newLeader = node;
                break;
            }
        }
        
        if (newLeader == null) {
            System.out.println("✗ 重新选举失败！");
            return;
        }
        
        System.out.println("\n第四阶段：新领导者继续工作");
        System.out.println("新领导者: " + newLeader.getNodeId() + ", 任期: " + newLeader.getCurrentTerm());
        
        // 新leader添加更多日志
        for (int i = 4; i <= 6; i++) {
            newLeader.appendEntry("New-Leader-Log-" + i);
            Thread.sleep(200);
        }
        
        printClusterStatus(nodes);
        
        // 故障leader重新加入
        System.out.println("\n第五阶段：故障领导者重新加入集群");
        System.out.println("故障领导者 " + failedLeaderId + " 尝试重新加入...");
        
        // 重新启动故障的leader（模拟网络恢复）
        initialLeader.rejoinCluster();
        
        Thread.sleep(1000); // 等待重新加入的处理
        
        System.out.println("重新加入后的集群状态:");
        printClusterStatus(nodes);
        
        // 第六阶段：模拟任期冲突场景
        System.out.println("\n第六阶段：模拟两个相同任期Leader的冲突场景");
        
        // 找到两个不同的节点来模拟冲突
        RaftNode conflictNode1 = null;
        RaftNode conflictNode2 = null;
        
        for (RaftNode node : nodes) {
            if (node.getState() != RaftState.LEADER) {
                if (conflictNode1 == null) {
                    conflictNode1 = node;
                } else if (conflictNode2 == null) {
                    conflictNode2 = node;
                    break;
                }
            }
        }
        
        if (conflictNode1 != null && conflictNode2 != null) {
            int conflictTerm = newLeader.getCurrentTerm() + 1;
            
            System.out.println("创建冲突场景：");
            System.out.println("- 强制设置 " + conflictNode1.getNodeId() + " 为Leader，任期: " + conflictTerm);
            System.out.println("- 强制设置 " + conflictNode2.getNodeId() + " 为Leader，任期: " + conflictTerm);
            
            // 强制设置两个节点为相同任期的leader（模拟网络分区后的脑裂）
            conflictNode1.forceSetLeader(conflictTerm);
            Thread.sleep(100);
            conflictNode2.forceSetLeader(conflictTerm);
            
            System.out.println("\n冲突创建后的集群状态:");
            printClusterStatus(nodes);
            
            // 观察冲突解决过程
            System.out.println("\n第七阶段：观察Raft冲突解决机制");
            System.out.println("等待Raft算法自动解决任期冲突...");
            
            // 让冲突的leader尝试发送心跳，触发任期比较
            Thread.sleep(3000);
            
            System.out.println("\n冲突解决后的集群状态:");
            printClusterStatus(nodes);
            
            // 验证最终状态
            verifyFinalState(nodes);
            
        } else {
            System.out.println("无法找到足够的节点来模拟冲突场景");
        }
        
        // 清理资源
        System.out.println("\n清理资源...");
        for (RaftNode node : nodes) {
            try {
                node.shutdown();
                InMemoryMessageHandler.unregisterNode(node.getNodeId());
            } catch (Exception e) {
                // 节点可能已经关闭
            }
        }
        
        System.out.println("复杂故障转移场景演示完成！");
    }
    
    /**
     * 验证最终集群状态
     */
    private static void verifyFinalState(RaftNode[] nodes) {
        System.out.println("\n=== 最终状态验证 ===");
        
        int leaderCount = 0;
        int maxTerm = 0;
        RaftNode finalLeader = null;
        
        for (RaftNode node : nodes) {
            if (node.getState() == RaftState.LEADER) {
                leaderCount++;
                finalLeader = node;
            }
            maxTerm = Math.max(maxTerm, node.getCurrentTerm());
        }
        
        System.out.println("领导者数量: " + leaderCount);
        System.out.println("最高任期: " + maxTerm);
        
        if (leaderCount == 1) {
            System.out.println("✓ 集群状态正常，唯一领导者: " + finalLeader.getNodeId());
            
            // 测试最终leader是否能正常工作
            System.out.println("测试最终领导者功能...");
            finalLeader.appendEntry("Final-Test-Log");
            
            try {
                Thread.sleep(500);
                System.out.println("✓ 最终领导者工作正常");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            
        } else if (leaderCount == 0) {
            System.out.println("⚠ 当前没有领导者，可能正在选举中");
        } else {
            System.out.println("✗ 仍然存在多个领导者，需要更多时间解决冲突");
        }
        
        // 检查任期一致性
        boolean termConsistent = true;
        for (int i = 1; i < nodes.length; i++) {
            if (nodes[i].getCurrentTerm() != nodes[0].getCurrentTerm()) {
                termConsistent = false;
                break;
            }
        }
        
        if (termConsistent) {
            System.out.println("✓ 所有节点任期一致");
        } else {
            System.out.println("⚠ 节点任期不一致，仍在同步中");
        }
    }
} 